#include <iostream> 
#include <fstream>
#include <sstream>
#include <vector>
#include <list>
#include <pthread.h>

#include "polimr.hpp"
#include "Partitioner.hpp"
#include "IntermBuffer.hpp"
#include "ResultBuffer.hpp"
#include "Mapper.hpp"
#include "Reducer.hpp"

using namespace std; 

ostream& operator<< (ostream& s, Partition p)
{
  return s << p.getBuffer() << "\n"; 
}

ostream& operator<< (ostream& s, Partition *p)
{
  return s << p->getBuffer() << "\n";   
}

typedef struct reduce_arg
{
  int r; 
} rarg_t; 

typedef int (*f_hash) (string& key); 

namespace MapReduce
{
  static int M, R; 
  static Partitioner* parts; 
  static pthread_mutex_t parts_lock;
  static IntermBuffer* intermbuf;
  static ResultBuffer* results; 

  static f_hash rslot; 

  static int default_hash(string& key)
  {
    unsigned long hash = 5381; 

    for (int unsigned i = 0; i < key.size(); i++)
      hash = ((hash << 5) + (int unsigned)key[i]); 
    return hash % R; 
  }

  void emit_intermediate(string key, string value)
  {
    int slot = rslot(key); 
    intermbuf->addEntry(slot, pair<string, string>(key, value)); 
  }

  void emit_final(int r, string key, string finval)
  {
    results->addEntry(r, pair<string, string>(key, finval)); 
  }

  static void* mapper_run(void *params)
  {
    bool finished = false; 
    Partition* input; 
    Mapper* mapper = createMapper(); 

    while (!finished)
      {
	pthread_mutex_lock(&parts_lock);
	if (parts->hasPartitions())
	  input = parts->getPartition(); 
	else
	  {
	    finished = true;
	    input = NULL; 
	  }
	pthread_mutex_unlock(&parts_lock);
	
	// pass input to user map function
	if (input != NULL)
	  mapper->map(input->getBuffer()); 
      }
    return NULL; 
  }

  static void *reducer_run(void *arg)
  {
    rarg_t *rarg = (rarg_t *)arg; 
    int r = rarg->r;
    Reducer* reducer = createReducer(); 
    reducer->setRid(r); 

    list<pair<string, string> >& data = intermbuf->getBuffer(r); 
    data.sort(); 
    
    list<pair<string, string> >::const_iterator it;
    list<string> values; 
    string crtkey = data.front().first;
    reducer->setKey(crtkey); 
    for (it = data.begin(); it != data.end(); it++)
      {
	if (it->first != crtkey)
	  {
	    reducer->reduce(values); 
	    crtkey = it->first;
	    reducer->setKey(crtkey); 
	    values.clear();
	  }
	values.push_back(it->second);
      }
    
    if (values.size() > 0)
      reducer->reduce(values); 

    delete rarg; 
    return NULL; 
  }

  static void spawn_mappers()
  {
    pthread_t *mappers = new pthread_t[M]; 
    
    pthread_mutex_init(&parts_lock, NULL); 
    // spawn mappers 
    for (int m = 0; m < M; m++)
      if (pthread_create(&mappers[m], NULL, &mapper_run, NULL))
	{
	  perror("pthread_create for mapper"); 
	  exit(EXIT_FAILURE); 
	}
    
    // wait mappers to finish 
    for (int m = 0; m < M; m++)
      if (pthread_join(mappers[m], NULL))
	{
	  perror("pthread_join for mapper"); 
	  exit(EXIT_FAILURE); 
	}
  }

  static void spawn_reducers()
  {
    pthread_t *reducers = new pthread_t[R]; 

    // spawn reducers
    for (int r = 0; r < R; r++)
      {
	rarg_t *arg = new rarg_t; 
	arg->r = r; 
	if (pthread_create(&reducers[r], NULL, &reducer_run, arg))
	  {
	    perror("pthread_create for reducer"); 
	    exit(EXIT_FAILURE); 
	  }
      }
    // wait mappers to finish 
    for (int r = 0; r < R; r++)
      if (pthread_join(reducers[r], NULL))
	{
	  perror("pthread_join for reducer"); 
	  exit(EXIT_FAILURE); 
	}
  }

  static void init(mrconfig& params)
  {
    M = params.M; 
    R = params.R; 
    rslot = default_hash; 
    parts = new Partitioner(params.input); 
    intermbuf = new IntermBuffer(R);
    results = new ResultBuffer(R, params.outname); 
  }
  
  static void free_resources()
  {
    delete parts; 
    delete intermbuf;
    delete results; 
  }
  
  int mapreduce(mrconfig& params)
  {
    init(params); 
    spawn_mappers(); 
    spawn_reducers();
    free_resources(); 
    return 0; 
  }
}


